package bing.search.hbase;

import java.io.IOException;
import java.util.List;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.CellUtil;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.HColumnDescriptor;
import org.apache.hadoop.hbase.HTableDescriptor;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.Admin;
import org.apache.hadoop.hbase.client.Connection;
import org.apache.hadoop.hbase.client.ConnectionFactory;
import org.apache.hadoop.hbase.client.Delete;
import org.apache.hadoop.hbase.client.Get;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.client.ResultScanner;
import org.apache.hadoop.hbase.client.Scan;
import org.apache.hadoop.hbase.client.Table;
import org.apache.hadoop.hbase.filter.CompareFilter.CompareOp;

import com.google.common.collect.Lists;

import bing.search.common.Document;

import org.apache.hadoop.hbase.filter.RegexStringComparator;
import org.apache.hadoop.hbase.filter.RowFilter;

public class HbaseUtils {

	// hbase 表名
	public static final TableName DOCUMENT_TABLE_NAME = TableName.valueOf("doc");
	// hbase 表列簇 ,文章信息
	public final String DOCUMENT_TABLE_CF_1 = "cf";

	// 列簇1中的列
	public final String COLUMNFAMILY_1_TITLE = "title";
	public final String COLUMNFAMILY_1_AUTHOR = "author";
	public final String COLUMNFAMILY_1_CONTENT = "content";
	public final String COLUMNFAMILY_1_DESCRIBE = "describe";

	public static Admin admin = null;
	public static Configuration conf = null;
	public static Connection conn = null;

	// 构造函数加载配置
	public HbaseUtils() {
		conf = HBaseConfiguration.create();
		conf.set("hbase.zookeeper.quorum", "lbh3,lbh4,lbh5");

		try {
			conn = ConnectionFactory.createConnection(conf);
			admin = conn.getAdmin();
		} catch (IOException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
	}

	public static void main(String[] args) {
		// TODO Auto-generated method stub
	}

	// rowFilter 的使用
	public void getRowFilter(TableName tableName, String reg) throws IOException {
		Table table = conn.getTable(tableName);
		// Filter
		RowFilter rowFilter = new RowFilter(CompareOp.NOT_EQUAL, new RegexStringComparator(reg));
		Scan scan = new Scan();
		scan.setFilter(rowFilter);
		ResultScanner scanner = table.getScanner(scan);

		for (Result result : scanner) {
			System.out.println(String.valueOf(result.getRow()));
		}
	}

	public void getScanData(TableName tableName, String family, String qualifier) throws IOException {
		Table table = conn.getTable(tableName);
		Scan scan = new Scan();
		scan.addColumn(family.getBytes(), qualifier.getBytes());

		ResultScanner scanner = table.getScanner(scan);
		for (Result result : scanner) {
			if (result.rawCells().length == 0) {
				System.out.println(tableName.getNameAsString() + " 表数据为空.");
			} else {
				for (Cell cell : result.rawCells()) {
					System.out.println(new String(CellUtil.cloneRow(cell)) + "\t" + new String(CellUtil.cloneValue(cell)));
				}
			} // if-else
		} // for

	}

	/*删除数据表*/
	public void deleteTable(TableName tableName) {
		try {
			if (admin.tableExists(tableName)) {
				admin.disableTable(tableName);
				admin.deleteTable(tableName);
				System.out.println(tableName.getNameAsString()+" 表删除成功");
			}
		} catch (IOException e) {
			e.printStackTrace();
			System.out.println(tableName.getNameAsString()+" 表删除失败");
		}
	}

	/*
	 * 删除一条记录
	 */
	public void deleteOneRecord(TableName tableName,String rowKey) {
		try {
			Table table=conn.getTable(tableName);
			Delete delete=new Delete(rowKey.getBytes());
			table.delete(delete);
			
			System.out.println(rowKey+" 记录删除成功");
		} catch (IOException e) {
			e.printStackTrace();
			System.out.println(rowKey+" 记录删除失败");
		}
	}
	
	/*
	 * 获取表中所有数据
	 */
	public void getAllData(TableName tableName) {
		try {
			Table table=conn.getTable(tableName);
			Scan scan=new Scan();
			
			ResultScanner scanner=table.getScanner(scan);
			for(Result result:scanner) {
				if(result.rawCells().length==0) {
					System.out.println(tableName.getNameAsString() + " 表数据为空.");
				} else {
					for(Cell cell:result.rawCells()) {
						System.out.println(new String(CellUtil.cloneRow(cell)) + "\t" + new String(CellUtil.cloneValue(cell)));
					}
				}
			}
		} catch (IOException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
	}
	
	/*读取一条记录*/
	public Document get(TableName tableName, String row) throws IOException {
		Table table =conn.getTable(tableName);
		Get get=new Get(row.getBytes());
		
		Document doc=null;
		
		Result result=table.get(get);
		Cell[] cells=result.rawCells();
		if(cells.length==4) {
			doc=new Document();
			doc.setId(Integer.parseInt(row));
			doc.setTitle(String.valueOf(CellUtil.cloneValue(cells[3])));
			doc.setAuthor(String.valueOf(CellUtil.cloneValue(cells[0])));
			doc.setContent(String.valueOf(CellUtil.cloneValue(cells[1])));
			doc.setDescribe(String.valueOf(CellUtil.cloneValue(cells[2])));
		}
		return doc;
	}
	
	/*添加一条记录*/
	public void put(TableName tableName, String row, String family,String column,String data) throws IOException {
		if(!admin.tableExists(tableName)) {
			System.out.println("table no exist.");
			this.createTable(tableName, family);
		}
		Table table=conn.getTable(tableName);
		Put put=new Put(row.getBytes());
		put.addColumn(family.getBytes(), column.getBytes(), data.getBytes());
		
		table.put(put);
		
		System.out.println("put '"+row+"',"+family+":"+column+",'"+data+"'");
	}
	
	/*查询所有表名*/
	public List<TableName> getAllTable() throws IOException {
		List<TableName> tables=Lists.newArrayList();
		if(admin!=null) {
			HTableDescriptor[] listTables=admin.listTables();
			for(HTableDescriptor table :listTables) {
				tables.add(table.getTableName());
				System.out.println(table.getNameAsString());
			}
		}
		return tables;
	}
	
	public void createTable(TableName tableName, String column) throws IOException {
		if(admin.tableExists(tableName)) {
			System.out.println(tableName.getNameAsString()+" 表已经存在");
		} else {
			HTableDescriptor table=new HTableDescriptor(tableName);
			table.addFamily(new HColumnDescriptor(column.getBytes()));
			admin.createTable(table);
			System.out.println(tableName.getNameAsString()+" 表创建成功");
		}
	}
}
